package com.galeno.blink.demo;

import com.galeno.FlinkUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

import java.util.Random;

/**
 * @author galeno
 * @Title:
 * @Description:
 * @date 2021/10/1921:05
 */
public class Connect {
    public static void main(String[] args) throws Exception {
        //Random r = new Random();
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8888);
        StreamExecutionEnvironment locEnvWebUI = FlinkUtils.getLocEnvWebUI(configuration);
        DataStreamSource<String> lines1 = locEnvWebUI.socketTextStream("192.168.77.101", 8888);
        DataStreamSource<String> lines2 = locEnvWebUI.socketTextStream("192.168.77.101", 9999);
        ConnectedStreams<String, String> connect = lines1.connect(lines2);
        SingleOutputStreamOperator<String> map = connect.map(new CoMapFunction<String, String, String>() {

            final Random r = new Random();
            @Override
            public String map1(String s) throws Exception {
                return s+r.nextInt(100);
            }

            @Override
            public String map2(String s) throws Exception {
                return s+r.nextInt(50);
            }
        });


        map.print();
        locEnvWebUI.execute();


    }
}
